package com.hexiang.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;

@Component
@Slf4j
public class DemoMQProducer {

    private final String TOPIC = "test_topic";

    private DefaultMQProducer defaultMQProducer;

    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer messageTimeOut;


    @PostConstruct
    public void start() throws MQClientException {
        defaultMQProducer = new DefaultMQProducer(producerGroup, getAclRPCHook());
        defaultMQProducer.setNamesrvAddr(nameServer);
        defaultMQProducer.start();
    }

    @PreDestroy
    public void shutdown() {
        defaultMQProducer.shutdown();
    }

    RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("rocketmq_u1", "12345678"));
    }

    /**
     * 发送普通消息
     */
    public void sendMsg(String msgBody) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(TOPIC, (msgBody).getBytes(StandardCharsets.UTF_8));
        defaultMQProducer.setRetryTimesWhenSendFailed(1);
        SendResult sendResult = defaultMQProducer.send(message);
        log.warn("sendResult " + sendResult);
    }

    /**
     * 发送异步消息 在SendCallback中可处理相关成功失败时的逻辑
     */
    public void sendAsyncMsg(String msgBody) throws RemotingException, MQClientException, InterruptedException {
        Message message = new Message(TOPIC, (msgBody).getBytes(StandardCharsets.UTF_8));


        defaultMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 处理消息发送成功逻辑
                log.warn("sendResult " + sendResult);

            }

            @Override
            public void onException(Throwable e) {
                // 处理消息发送异常逻辑
                log.error("sendResultErr ", e);

            }
        });
    }

    /**
     * 发送单向消息
     *
     * @param msgBody
     * @throws UnsupportedEncodingException
     * @throws RemotingException
     * @throws MQClientException
     * @throws InterruptedException
     */
    public void sendOneway(String msgBody) throws RemotingException, MQClientException, InterruptedException {
        Message message = new Message(TOPIC, (msgBody).getBytes(StandardCharsets.UTF_8));

        defaultMQProducer.sendOneway(message);
    }

    /**
     * 发送延时消息<br/>
     * 在start版本中 延时消息一共分为18个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h<br/>
     */
    public void sendDelayMsg(String msgBody, Integer delayLevel) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(TOPIC, (msgBody).getBytes(StandardCharsets.UTF_8));
        message.setDelayTimeLevel(delayLevel);
        SendResult sendResult = defaultMQProducer.send(message, messageTimeOut);

        log.warn("sendResult " + sendResult.getMsgId());

    }

    /**
     * 发送带tag的消息,直接在topic后面加上":tag"
     */
    public void sendTagMsg(String msgBody) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(TOPIC, "tag1", (msgBody).getBytes(StandardCharsets.UTF_8));

        SendResult sendResult = defaultMQProducer.send(message);
        log.warn("sendResult " + sendResult.getMsgId());

    }

}